using System;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Reactive.Disposables;
using System.Reactive.Linq;
#if DEBUG
using System.Threading;
#endif
using System.Threading.Tasks;
using Microsoft.Owin;

namespace LinqToOwin
{
  public sealed class LinqToOwinMiddleware : OwinMiddleware
  {
    private readonly CompositeDisposable subscriptions;
    private readonly Func<LinqToOwinHost, IEnumerable<IObservable<object>>> queryFactory;

    public LinqToOwinMiddleware(OwinMiddleware next, CompositeDisposable subscriptions, Func<LinqToOwinHost, IEnumerable<IObservable<object>>> queryFactory)
      : base(next)
    {
      Contract.Requires(subscriptions != null);
      Contract.Requires(queryFactory != null);

      this.subscriptions = subscriptions;
      this.queryFactory = queryFactory;
    }

    [ContractInvariantMethod]
    [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Performance", "CA1822:MarkMembersAsStatic", Justification = "Required for code contracts.")]
    private void ObjectInvariant()
    {
      Contract.Invariant(subscriptions != null);
      Contract.Invariant(queryFactory != null);
    }

    public override async Task Invoke(IOwinContext context)
    {
      Contract.Assume(context != null);

      var completed = new TaskCompletionSource<bool>();

      try
      {
#if DEBUG
        context.TraceOutput.WriteLine("LINQ to Owin received \"{0}\" request for \"{1}\".", context.Request.Method, context.Request.Path);
#endif

        // Create a new query graph per request to ensure thread-safety, since Rx operators don't support concurrent notifications. (4.2 Rx Design Guidelines)
        var wrapper = new LinqToOwinContext(context, completed.Task);
        var queryContext = new LinqToOwinHost(Observable.Return(wrapper));
        var queries = queryFactory(queryContext);

        await ExecuteSequentialAsync(wrapper, queries).ConfigureAwait(false);

        completed.TrySetResult(true);
      }
      catch (Exception ex)
      {
        completed.TrySetException(ex);
        throw;
      }
    }

    private async Task ExecuteSequentialAsync(LinqToOwinContext context, IEnumerable<IObservable<object>> queries)
    {
      Contract.Requires(context != null);
      Contract.Requires(queries != null);

      var resetHandled = context.SetHandledUponNotification;

      int i = 0;
      foreach (var query in queries)
      {
        i++;

#if DEBUG
        context.TraceOutput.WriteLine("Before execute query {0} on thread #{1}; Context: {2}; Handled: {3}", i, Thread.CurrentThread.ManagedThreadId, context.GetHashCode(), context.Handled);
#endif

        Contract.Assume(context.SetHandledUponNotification);

        var cap = new QueryCap(context);
        var subscription = new SingleAssignmentDisposable();

        subscriptions.Add(subscription);
        subscription.Disposable = query.Finally(() => subscriptions.Remove(subscription)).Subscribe(cap);

        await cap.Task.ConfigureAwait(false);

#if DEBUG
        context.TraceOutput.WriteLine("After execute query {0} on thread #{1}; Context: {2}; Handled: {3}", i, Thread.CurrentThread.ManagedThreadId, context.GetHashCode(), context.Handled);
#endif

        context.SetHandledUponNotification = resetHandled;
      }
    }
  }
}